package com.dstation.jobhandle;

import cn.hutool.core.text.csv.CsvUtil;
import cn.hutool.core.text.csv.CsvWriter;
import cn.hutool.core.util.CharsetUtil;
import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.recommendation.AlsTrainBatchOp;
import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.alibaba.alink.pipeline.recommendation.AlsItemsPerUserRecommender;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.dstation.utils.RedisPreKey;
import org.apache.flink.types.Row;
import com.dstation.domain.RecommendInfo;
import com.dstation.service.interf.IRecommendService;
import com.dstation.utils.RecommendStrategyUtil;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;

import org.springframework.stereotype.Component;
import redis.clients.jedis.*;

import javax.annotation.Resource;
import java.io.File;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Component
@Slf4j
public class AlsOffLineRecommedJob {
    static String REGEX_SPACE = "\r|\n";//定义空格回车换行符
    static Pattern P_SPACE = Pattern.compile(REGEX_SPACE, Pattern.CASE_INSENSITIVE);


    @Resource
    private IRecommendService recommendService;


    @Resource
    private JedisPool redisPoolFactory;


    @XxlJob("alsOffLineRecommedJob")
    public ReturnT<String> alsOffLineRecommedJob(String param) throws Exception {
        XxlJobLogger.log("开始---->推荐  离线数据部分 {}", param);

        log.info("开始---->推荐  离线数据部分");

        //对遗留数据删除
        removeAllRecentRecommendData();
        delStrings(RedisPreKey.ALS_OFFLINE_REC_DATA);


        int dataSize= RecommendStrategyUtil.RECOMMEND_DATA_SIZE;
        try{
            //先处理浏览数据部分，全量处理
            calculateBrowserData();

//            点赞数据部分
            calculateLikedData();
//            收藏数据部分
            calculateCollectData();

//          硬币数据部分
            calculateCoinData();

            String fileName="rating";

            final File ratingCsvFile = File.createTempFile(fileName, ".csv");//创建临时文件
            log.info("临时文件所在的本地路径：" + ratingCsvFile.getCanonicalPath());


            //指定路径和编码
            CsvWriter writer = CsvUtil.getWriter(ratingCsvFile, CharsetUtil.CHARSET_UTF_8);

            //处理模块数据
            List<RecommendInfo> results = recommendService.findAllRecommend();
            writer.writeBeans(results);


            String filePath = ratingCsvFile.getAbsolutePath();
            String schema = " userId int, videoId int, rating double";
            CsvSourceBatchOp csvSource = new CsvSourceBatchOp().setIgnoreFirstLine(true).setFilePath(filePath)
                    .setSchemaStr(schema).setFieldDelimiter(",");

            BatchOperator<?> als = new AlsTrainBatchOp().setUserCol("userId").setItemCol("videoId").setRateCol("rating")
                    .setNumIter(10).setRank(10).setLambda(0.01);
            BatchOperator<?> model = als.linkFrom(csvSource);


            AlsItemsPerUserRecommender alsRec = new AlsItemsPerUserRecommender().setUserCol("userId").setRecommCol("videoId")
                    .setK(dataSize).setReservedCols("userId").setModelData(model);


            BatchOperator<?> transResult = alsRec.transform(csvSource);



            Jedis jedis = redisPoolFactory.getResource();
            Pipeline pipline = jedis.pipelined();
            transResult.distinct().lazyCollect(new Consumer<List<Row>>() {
                @Override
                public void accept(List<Row> d) {
                    for (Row row : d) {

//				  System.out.println(row.toString()); //redis 为每个用户存储

                        JSONObject obj = JSON.parseObject(row.getFieldAs(1));
                        JSONArray ars = obj.getJSONArray("videoId");
                        JSONArray scs = obj.getJSONArray("rate");
                        int source = 1;
                        //拼装redis的key
                        StringBuilder keyName = new StringBuilder(RedisPreKey.ALS_OFFLINE_REC_DATA);
                        keyName.append(row.getFieldAs(0).toString());
                        log.info(keyName.toString());
                        Map<String, Double> map = new HashMap();

                        for (int i = 0; i < ars.size() ; i++) {
                            //装载个性化的模块数据


                            map.put(String.valueOf(ars.getIntValue(i)), (double)i);

                        }
//                        对各个用户的不同模块数据做批量插入
                        pipline.zadd(keyName.toString(), map);
                    }

                }
            });

            BatchOperator.execute();

            pipline.sync();


            //删除临时cvs文件
            ratingCsvFile.delete();
        }catch (Exception e){
            log.info("离线推荐数据部分出错！",e.getMessage());
            XxlJobLogger.log("离线推荐数据部分出错！",e.getMessage());

            e.printStackTrace();
        }

        XxlJobLogger.log("结束---->推荐  离线数据部分 {}", param);
        log.info("结束 ---->推荐  离线数据部分");


        return ReturnT.SUCCESS;
    }

    private void removeAllRecentRecommendData() {
        boolean isSuccess=recommendService.dropAllRecommendData();

        XxlJobLogger.log("推荐     初始化：｛｝",isSuccess);
        log.info("推荐      初始化：｛｝",isSuccess);
    }


    private void calculateBrowserData() {
        boolean isSuccess=recommendService.addBrowserData();
        
        XxlJobLogger.log("推荐      浏览数据转换：｛｝",isSuccess);
        log.info("推荐      浏览数据转换：｛｝",isSuccess);
    }
    private void calculateCollectData() {
        boolean isSuccess=recommendService.addCollectData();
        XxlJobLogger.log("推荐      收藏数据转换：｛｝",isSuccess);
        log.info("推荐      收藏数据转换：｛｝",isSuccess);
    }

    private void calculateLikedData() {
        boolean isSuccess=recommendService.addLikedData();
        XxlJobLogger.log("推荐      点赞数据转换：｛｝",isSuccess);
        log.info("推荐     点赞数据转换：｛｝",isSuccess);
    }

    private void calculateCoinData() {
        boolean isSuccess=recommendService.addCoinData();
        XxlJobLogger.log("推荐      投币数据转换：｛｝",isSuccess);
        log.info("推荐     投币数据转换：｛｝",isSuccess);
    }

    /**
     * @param str
     * @return 过滤空格
     */
    public static String trimSpaceTag(String str) {
        Matcher m_space = P_SPACE.matcher(str);
        str = m_space.replaceAll(""); // 过滤空格回车标签
        return str.trim(); // 返回文本字符串
    }

    public void delStrings(String key) {

        try {
                 Jedis jedis = redisPoolFactory.getResource();
                 // 存入键值对
                 ScanParams scanParams = new ScanParams();
                 StringBuilder paramKey = new StringBuilder(key).append("*");
                 scanParams.match(paramKey.toString());
                 scanParams.count(1000);
                 ScanResult<String> sr = jedis.scan("0", scanParams);
                 List<String> a = sr.getResult();
            XxlJobLogger.log("本次删除共有"+a.size());
                 for (String delkey : a) {
                         jedis.del(delkey);
                     }
                 jedis.close();
             } catch (Exception e) {
                    e.printStackTrace();
             }
     }
}
